use std::{
    collections::HashMap,
    ops::Add,
    sync::Arc,
    time::{Duration, Instant, SystemTime, UNIX_EPOCH},
};

use anyhow::Error;
use async_trait::async_trait;
use futures::{lock::Mutex, Future};
use log::{debug, error, warn};
use russh::{
    client::{Config, Handle, Handler, Session},
    ChannelId,
};
use russh_keys::key;
use tokio::{
    sync::mpsc::{Receiver, Sender},
    time::sleep,
};
use yang2::context::Context;

use crate::{
    codec::XmlNcMsgCodec, Capabilities, NcMsg, NcMsgCodec, SourceType, XmlEncodeAble,
    NC_TRUNK_FRAG_SIZE,
};

// 默认读取消息超时时间
const READ_TIMEOUT: u64 = 5000;

pub trait NetconfClient {
    /// get running data
    fn get(&self, filter: Option<&str>) -> impl Future<Output = Result<NcMsg, Error>>;

    /// get running data
    fn get_xpath(&self, xpath: &str) -> impl Future<Output = Result<NcMsg, Error>>;

    /// get config data
    fn get_config(
        &self,
        target: &SourceType,
        filter: Option<&str>,
    ) -> impl Future<Output = Result<NcMsg, Error>>;

    /// get config data
    fn get_config_xpath(
        &self,
        target: &SourceType,
        xpath: &str,
    ) -> impl Future<Output = Result<NcMsg, Error>>;

    /// edit config data
    fn edit_config(
        &self,
        target: &SourceType,
        content: &str,
    ) -> impl Future<Output = Result<NcMsg, Error>>;

    /// call rpc
    fn call_rpc(&self, content: &str) -> impl Future<Output = Result<NcMsg, Error>>;

    /// get capabilities
    fn capabilities(&self) -> Result<Capabilities, Error>;
}

#[derive(Debug, Clone)]
pub struct Client {
    addr: (String, u16),
    username: String,
    password: String,
    notify_sender: Sender<NcMsg>,
    codec: XmlNcMsgCodec,
    pub data_cache: Vec<u8>,
    tx: Sender<String>,
    msgs: Arc<Mutex<HashMap<String, NcMsg>>>,
}

impl Client {
    pub fn new(
        addr: (&str, u16),
        username: &str,
        password: &str,
        tx: Sender<String>,
        notify_sender: Sender<NcMsg>,
        ctx: Arc<Context>,
    ) -> Client {
        Client {
            addr: (addr.0.to_string(), addr.1),
            username: username.to_string(),
            password: password.to_string(),
            notify_sender,
            codec: XmlNcMsgCodec { ctx },
            data_cache: vec![],
            tx,
            msgs: Arc::new(Mutex::new(HashMap::new())),
        }
    }

    pub async fn connect(&mut self, mut rx: Receiver<String>) -> Result<Handle<Client>, Error> {
        let mut session = russh::client::connect(
            Arc::new(Config::default()),
            (self.addr.0.as_str(), self.addr.1),
            self.clone(),
        )
        .await?;
        session
            .authenticate_password(&self.username, &self.password)
            .await?;

        let channel = session.channel_open_session().await?;
        tokio::spawn(async move {
            loop {
                if let Some(data) = rx.recv().await {
                    let data = data.as_bytes();

                    let dlen = data.len();
                    let mut chrunk_pos = 0;
                    while chrunk_pos < dlen {
                        if chrunk_pos + NC_TRUNK_FRAG_SIZE > dlen {
                            let chk_len = dlen - chrunk_pos;
                            let mut chk_bytes: Vec<u8> = vec![0x0A, 0x23];
                            chk_bytes.append(&mut chk_len.to_string().as_bytes().to_vec());
                            chk_bytes.append(&mut vec![0x0A]);
                            chk_bytes.append(&mut data[chrunk_pos..chrunk_pos + chk_len].to_vec());
                            chk_bytes.append(&mut vec![0x0A, 0x23, 0x23, 0x0A]);
                            chrunk_pos += chk_len;
                            channel.data::<&[u8]>(&chk_bytes).await.unwrap();
                            debug!("send chunk data, len: {}", chk_bytes.len());
                        } else {
                            let chk_len = NC_TRUNK_FRAG_SIZE;
                            let mut chk_bytes: Vec<u8> = vec![0x0A, 0x23];
                            chk_bytes.append(&mut chk_len.to_string().as_bytes().to_vec());
                            chk_bytes.append(&mut vec![0x0A]);
                            chk_bytes.append(&mut data[chrunk_pos..chrunk_pos + chk_len].to_vec());
                            chrunk_pos += chk_len;
                            channel.data::<&[u8]>(&chk_bytes).await.unwrap();
                            debug!("send chunk data, len: {}", chk_bytes.len());
                        }
                    }
                } else {
                    break;
                }
            }
        });

        Ok(session)
    }

    /// send a rpc message and wait for response by msg id until timeout
    async fn send_request(&self, content: &str) -> Result<NcMsg, Error> {
        let time = SystemTime::now()
            .duration_since(UNIX_EPOCH)
            .unwrap()
            .as_millis();
        let msg_id = time.to_string();
        self.tx
            .send(format!(
                r#"<rpc xmlns="urn:ietf:params:xml:ns:netconf:base:1.0" message-id="{}">{}</rpc>"#,
                msg_id, content
            ))
            .await
            .unwrap();
        let when_timeout = Instant::now().add(Duration::from_millis(READ_TIMEOUT));
        loop {
            let new_now = Instant::now();
            if when_timeout < new_now {
                return Err(Error::msg("timeout"));
            }
            if let Some(msg) = self.msgs.lock().await.remove(&msg_id) {
                return Ok(msg.clone());
            }
            sleep(Duration::from_millis(20)).await;
        }
    }
}

impl NetconfClient for Client {
    async fn get(&self, filter: Option<&str>) -> Result<NcMsg, Error> {
        match filter {
            Some(content) => {
                self.send_request(&format!(
                    r#"<get><filter xmlns:ne="coresat.com:yang:ne" type="subtree">{}</filter></get>"#,
                    content
                ))
                .await
            }
            None => {
                self.send_request(r#"<get><filter  xmlns:ne="coresat.com:yang:ne" type="subtree"></filter></get>"#)
                    .await
            }
        }
    }

    async fn get_config(&self, target: &SourceType, filter: Option<&str>) -> Result<NcMsg, Error> {
        match filter {
            Some(content) => {
                self.send_request(&format!(
                    r#"<get-config><source><{}/></source><filter  xmlns:ne="coresat.com:yang:ne" type="subtree">{}</filter></get-config>"#,target.to_string(),
                    content
                )).await
            }
            None => {
                self.send_request(r#"<get-config><filter  xmlns:ne="coresat.com:yang:ne" type="subtree"></filter></get-config>"#)
                    .await
            }
        }
    }

    async fn edit_config(&self, target: &SourceType, content: &str) -> Result<NcMsg, Error> {
        self.send_request(&format!(
            r#"<edit-config><target><{}/></target><config>{}</config></edit-config>"#,
            target.to_string(),
            content
        ))
        .await
    }

    fn capabilities(&self) -> Result<Capabilities, Error> {
        todo!()
    }

    async fn get_xpath(&self, xpath: &str) -> Result<NcMsg, Error> {
        self.send_request(&format!(
            r#"<get><filter xmlns:ne="coresat.com:yang:ne" type="xpath" select="{}"/></get>"#,
            xpath
        ))
        .await
    }

    async fn get_config_xpath(&self, target: &SourceType, xpath: &str) -> Result<NcMsg, Error> {
        self.send_request(&format!(
            r#"<get-config><source><{}/></source><filter xmlns:ne="coresat.com:yang:ne" type="xpath" select="{}"/></get-config>"#,
            target.to_string(),
            xpath
        ))
        .await
    }

    async fn call_rpc(&self, content: &str) -> Result<NcMsg, Error> {
        self.send_request(content).await
    }
}

impl SourceType {
    fn to_string(&self) -> &str {
        match self {
            SourceType::Running => "running",
            SourceType::Candidate => "candidate",
            SourceType::Startup => "startup",
        }
    }
}

#[async_trait]
impl Handler for Client {
    type Error = anyhow::Error;

    async fn check_server_key(
        &mut self,
        server_public_key: &key::PublicKey,
    ) -> Result<bool, Self::Error> {
        debug!("server public key: {:?}", server_public_key);
        Ok(true)
    }

    async fn data(
        &mut self,
        _channel: ChannelId,
        data: &[u8],
        _session: &mut Session,
    ) -> Result<(), Self::Error> {
        let mut content = data.to_vec();
        debug!("ssh data coming, len: {}", content.len());
        self.data_cache.append(&mut content);

        let (pos, msgs) = self.codec.decode(&self.data_cache);
        let (_, r) = self.data_cache.split_at(pos);
        self.data_cache = r.to_vec();
        {
            if let Some(msgs) = msgs {
                for msg in msgs {
                    debug!("recv msg: {:?}", msg);
                    if let NcMsg::InvelidMsg(_msg_id, reason) = &msg {
                        warn!(
                            "invalid nc msg coming, decode with None result, reason: {}",
                            reason
                        );
                        continue;
                    }
                    match &msg {
                        NcMsg::Hello(hello) => {
                            debug!("hello msg coming, {}", hello.encode_xml());
                        }
                        NcMsg::RpcReply(msg_id, ncopmsg) => {
                            debug!(
                                "reply data coming, msg_id: {:?}, data: {:?}",
                                msg_id, ncopmsg
                            );
                            self.msgs.lock().await.insert(msg_id.clone(), msg);
                        }
                        NcMsg::Notifycation(time, data) => {
                            debug!("notification coming, time: {:?}, data: {:?}", time, data);
                            self.notify_sender.send(msg).await.unwrap();
                        }
                        NcMsg::InvelidMsg(_msg_id, reson) => {
                            error!("invalid msg: {:?}", reson);
                        }
                        _ => {
                            error!("invalid msg: {:?}", msg);
                        }
                    }
                }
            }
        }
        Ok(())
    }
}
